package com.sunzm.dynamic_rule.mock_data;

import com.sunzm.dynamic_rule.pojo.LogBean;
import com.alibaba.fastjson.JSON;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.commons.lang3.RandomUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.HashMap;
import java.util.Properties;

/**
 * 模拟用户行为日志
 *
 * @author Administrator
 * @version 1.0
 * @date 2021-07-03 22:15
 * <p>
 * {
 * "account": "Vz54E9Ya",
 * "appId": "cn.doitedu.app1",
 * "appVersion": "3.4",
 * "carrier": "中国移动",
 * "deviceId": "WEISLD0235S0934OL",
 * "deviceType": "MI-6",
 * "ip": "24.93.136.175",
 * "latitude": 42.09287620431088,
 * "longitude": 79.42106825764643,
 * "netType": "WIFI",
 * "osName": "android",
 * "osVersion": "6.5",
 * "releaseChannel": "豌豆荚",
 * "resolution": "1024*768",
 * "sessionId": "SE18329583458",
 * "timeStamp": 1594534406220
 * "eventId": "productView",
 * "properties": {
 * "pageId": "646",
 * "productId": "157",
 * "refType": "4",
 * "refUrl": "805",
 * "title": "爱得堡 男靴中高帮马丁靴秋冬雪地靴 H1878 复古黄 40码",
 * "url": "https://item.jd.com/36506691363.html",
 * "utm_campain": "4",
 * "utm_loctype": "1",
 * "utm_source": "10"
 * }
 * <p>
 * }
 * <p>
 * <p>
 * kafka中要先创建好topic
 * bin/kafka-topics.sh --bootstrap-server 10.0.8.11:9092 --create --topic applog --partitions 3 --replication-factor 1
 * <p>
 * 创建完后，检查一下是否创建成功：
 * bin/kafka-topics.sh --bootstrap-server 10.0.8.11:9092 --list
 * <p>
 * 使用消费者命令行消费数据
 * //从上次的位置开始消费
 * bin/kafka-console-consumer.sh --bootstrap-server 10.0.8.11:9092 --topic applog --group g-test
 *
 * //从最新的数据开始消费
 * bin/kafka-console-consumer.sh --bootstrap-server 10.0.8.11:9092 --topic applog
 *
 * //从头开始消费
 * bin/kafka-console-consumer.sh --bootstrap-server 10.0.8.11:9092 --topic applog --from-beginning
 *
 * //模拟生成者往kakfa发送数据
 * bin/kafka-console-producer.sh --bootstrap-server 10.0.8.11:9092 --topic applog
 */
public class MockActionLog {
    public static void main(String[] args) {

        //数据要发送到的kafka的topic名称
        String sendTopic = "applog";

        Properties props = new Properties();
        props.setProperty(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, "82.156.210.70:9093");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 构造一个kafka生产者客户端
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(props);

        //定义事件类型的枚举类型数组
        String[] events = {"pageView", "addCart", "collect", "thumbUp", "share", "search", "adShow", "adClick", "fetchCoupon", "useCoupon", "register"};

        boolean isRunning = true;

        while (isRunning) {
            LogBean bean = new LogBean();
            // 决定人数，3个就可能有1000人
            String account = StringUtils.leftPad(RandomStringUtils.randomNumeric(2), 4, "0");
            //String account = "0027";
            bean.setAccount(account);

            String eventId = events[RandomUtils.nextInt(0, events.length)];
            bean.setEventId(eventId);
            bean.setTimeStamp(System.currentTimeMillis());
            bean.setAppId("cn.do");
            bean.setAppVersion("3.0");
            bean.setCarrier("中国移动");
            bean.setDeviceId(account);
            bean.setDeviceType("mi10");
            bean.setIp("152.33.68.90");
            bean.setLatitude(56.29385792);
            bean.setLongitude(120.329857234);
            bean.setSessionId("sessionid" + account);
            bean.setResolution("1024*2048");
            bean.setReleaseChannel("小米应用");
            bean.setNetType("wifi");
            bean.setOsName("android");
            bean.setOsVersion("9.0");

            HashMap<String, String> properties = new HashMap<>();
            // 页面数量可能达到10种
            properties.put("pageId", RandomUtils.nextInt(1, 11) + "");
            if (eventId.equals("addCart") || eventId.equals("collect") || eventId.equals("share") || eventId.equals("thumbUp")) {
                properties.put("productId", RandomUtils.nextInt(1, 11) + "");
            }
            if (eventId.equals("fetchCoupon") || eventId.equals("useCoupon")) {
                properties.put("couponId", RandomUtils.nextInt(1, 11) + "");
            }
            if (eventId.equals("adShow") || eventId.equals("adClick")) {
                properties.put("adId", RandomUtils.nextInt(1, 11) + "");
            }

            if (eventId.equals("search")) {
                properties.put("keyword", RandomStringUtils.randomNumeric(1));
            }

            bean.setProperties(properties);

            kafkaProducer.send(new ProducerRecord<String, String>(sendTopic, JSON.toJSONString(bean)));

            try {
                //为了防止数据产生过快，睡眠一下
                Thread.sleep(RandomUtils.nextInt(2000, 20000));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

        }

        if (!isRunning) {
            kafkaProducer.close();
        }

        System.exit(0);
    }
}
